package com.hzh.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.io.Source

object Demo23Broadcast {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("Demo23Broadcast")
    conf.setMaster("local")

    val sc = new SparkContext(conf)

    /**
     * 广播变量
     * 当算子内使用算子外的一个比较大的变量时，可以将这个变量广播出去，可以减少变量的副本数
     *
     */
      //读取学生表，以学号作为key构建一个map集合
    val studentMap: Map[String, String] = Source
      .fromFile("data/students.txt")
      .getLines()
      .toList
      .map(stu => {
        val id: String = stu.split(",")(0)
        (id, stu)
      })
      .toMap

    val scoreRDD: RDD[String] = sc.textFile("data/score.txt")

    /**
     * 关联学生表和分数表
     * 循环分数表，使用学号到学生表的map集合中查询学生的信息
     *
     */

    /**
     * 将Driver端的一个普通变量广播到Executor
     *
     */


    scoreRDD.map(sco=>{
      val id: String = sco.split(",")(0)
      //使用学号到学生表中获取学生的信息
      val studentInfo: String = studentMap.getOrElse(id, "默认值")
      (sco,studentInfo)
    })
      .foreach(println)
  }
}
